package com.seewo.dubbo.debugger.admin.service.impl;

import com.alibaba.dubbo.common.URL;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import com.seewo.dubbo.debugger.admin.common.Constants;
import com.seewo.dubbo.debugger.admin.common.RedisRegistry;
import com.seewo.dubbo.debugger.admin.common.ServiceDiscovery;
import com.seewo.dubbo.debugger.admin.common.ZookeeperRegistry;
import com.seewo.dubbo.debugger.admin.entity.Registry;
import com.seewo.dubbo.debugger.admin.entity.Service;
import com.seewo.dubbo.debugger.admin.exception.DubboDebuggerException;
import com.seewo.dubbo.debugger.admin.mapper.RegistryMapper;
import com.seewo.dubbo.debugger.admin.mapper.ServiceMapper;
import com.seewo.dubbo.debugger.admin.service.IRegistryService;
import com.seewo.dubbo.debugger.admin.utils.EhcacheTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.*;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author chenliangliang
 * @since 2018-07-29
 */
@Slf4j
@org.springframework.stereotype.Service
public class RegistryServiceImpl extends ServiceImpl<RegistryMapper, Registry> implements IRegistryService {

    @Autowired
    private EhcacheTemplate ehcacheTemplate;

    @Autowired
    private ThreadPoolTaskExecutor executor;

    @Autowired
    private ServiceMapper serviceMapper;




    @Override
    public Registry getServiceList(Registry registry) {
        final String protocol = registry.getCProtocol();
        final String address = registry.getCAddress();
        final String protocolFilter = registry.getProtocolFilter();
        ServiceDiscovery serviceDiscovery = getServiceDiscovery(protocol, address);
        List<List<URL>> serviceProviders;
        try {
            serviceProviders = serviceDiscovery.getEnableServiceProviders();
        } catch (Exception e) {
            ehcacheTemplate.cacheEvict(Constants.CACHE_NAME_SERVICE_DISCOVERY, Constants.CACHE_NAME_SERVICE_DISCOVERY + "_" + address + protocol);
            throw new DubboDebuggerException("注册中心地址不可用");
        }
        //缓存1分钟数据
        return ehcacheTemplate.cacheable(Constants.CACHE_NAME_SERVICE_LIST, address + protocol + Objects.hash(protocolFilter), () -> getServiceList0(registry, serviceProviders));
    }


    private Registry getServiceList0(Registry registry, List<List<URL>> serviceProviders) {
        String protocolFilter = registry.getProtocolFilter();
        List<Service> services = new ArrayList<>();
        registry.setServices(services);
        Set<String> serviceSet = new HashSet<>(64);
        Set<String> addressSet = new HashSet<>();
        Set<String> protocolSet = new HashSet<>();
        for (List<URL> urls : serviceProviders) {
            for (URL url : urls) {
                String prot = url.getProtocol();
                if (StringUtils.isNotEmpty(protocolFilter) && !StringUtils.equals(prot, protocolFilter)) {
                    continue;
                }
                String addr = url.getAddress();
                String application = url.getParameter("application");
                String key = addr + "|" + application;
                if (!serviceSet.contains(key)) {
                    serviceSet.add(key);
                    addressSet.add(addr);
                    protocolSet.add(prot);
                    Service service = new Service();
                    service.setCAddress(addr);
                    service.setCName(application);
                    service.setCProtocol(prot);
                    services.add(service);
                }
            }
        }
        if (services.isEmpty()) {
            return registry;
        }
        List<Service> insertList = new ArrayList<>();
        Registry reg = selectOne(new EntityWrapper<Registry>()
                .eq("c_address", registry.getCAddress())
                .eq("c_protocol", registry.getCProtocol()));
        if (reg == null) {
            registry.setCCreateTime(new Date());
            registry.setCUpdateTime(new Date());
            if (insert(registry)) {
                for (Service service : services) {
                    service.setCRegistryId(registry.getCId());
                    service.setCCreateTime(new Date());
                    service.setCUpdateTime(new Date());
                    insertList.add(service);
                }
            }
        } else {
            final Integer registryId = reg.getCId();
            registry.setCId(registryId);
            registry.setCName(reg.getCName());
            List<Service> serviceList = serviceMapper.selectList(new EntityWrapper<Service>()
                    .eq("c_registry_id", registryId)
                    .in("c_address", addressSet)
                    .in("c_protocol", protocolSet));
            Map<Service, Service> serviceMap = new HashMap<>(64);
            serviceList.forEach(s -> serviceMap.put(s, s));
            for (Service service : services) {
                service.setCRegistryId(registryId);
                if (serviceMap.containsKey(service)) {
                    Service ser = serviceMap.get(service);
                    service.setCDesc(ser.getCDesc());
                } else {
                    service.setCCreateTime(new Date());
                    service.setCUpdateTime(new Date());
                    insertList.add(service);
                }
            }
        }
        //异步更新t_service表
        executor.execute(() -> updateTableService(insertList));

        return registry;
    }

    private void updateTableService(List<Service> insertList) {
        int size = insertList.size();
        if (size > 0) {
            int res = serviceMapper.batchInsert(insertList);
            if (res != size) {
                log.warn("批量插入service异常: 需要插入：" + size + "条，实际插入：" + res + "条");
            }
        }
    }






    @Override
    public ServiceDiscovery getServiceDiscovery(String protocol, String address) {
        //缓存客户端1小时
        return ehcacheTemplate.cacheable(Constants.CACHE_NAME_SERVICE_DISCOVERY, Constants.CACHE_NAME_SERVICE_DISCOVERY + "_" + address + protocol, () -> {
            if ("zookeeper".equals(protocol)) {
                return new ZookeeperRegistry(address);
            } else if ("redis".equals(protocol)) {
                return new RedisRegistry(address);
            } else {
                throw new RuntimeException(protocol + " not support");
            }
        });
    }

}
